package dk.twu.analyzer.core;

import com.google.common.base.Stopwatch;
import com.google.common.collect.Multimap;
import dk.twu.analyzer.loader.ITDDataLoader;
import dk.twu.analyzer.model.*;
import dk.twu.analyzer.util.AnalyzerConfig;
import dk.twu.analyzer.util.VectorUtil;
import org.apache.commons.math3.stat.correlation.PearsonsCorrelation;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.time.LocalDate;
import java.time.LocalDateTime;
import java.time.temporal.ChronoUnit;
import java.util.*;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
import java.util.stream.Stream;

public class IntraDayAnalyzer {
    private static Logger logger = LoggerFactory.getLogger(IntraDayAnalyzer.class);
    private final ExchangeList exchangeList;
    private final SymbolList symbolList;
    private final ITDDataLoader itdDataLoader;

    public IntraDayAnalyzer(AnalyzerConfig analyzerConfig) {
        exchangeList = ExchangeList.setupExchangeList(analyzerConfig);
        symbolList = SymbolList.setupSymbolList(analyzerConfig, exchangeList.getAllExchange());
        itdDataLoader = new ITDDataLoader(analyzerConfig);
    }

    public Collection<MatchEntry> getDayMatchEntry(IntraDayInput input) {
        List<String> exchanges = input.getExchanges();
        List<String> symbolKeys = input.getSymbolKeys();
        LocalDate startDate = input.getStartDate();
        int windowSize = input.getWindowSize();
        int dayRange = input.getDayRange();
        MarketDuration marketDurationUS = new MarketDuration(input.getMarketOpen(), input.getMarketClose(),
                ChronoUnit.MINUTES);
        Stream<MatchEntry> entryStream = Stream.empty();
        //Initialize windows
        for (int plusDays = 0; plusDays < dayRange; plusDays++) {
            LocalDate curDate = startDate.plusDays(plusDays);
            List<Window> windows = WindowCreator.createSizedWindow(curDate, marketDurationUS, ChronoUnit.MINUTES,
                    windowSize);
            Multimap<String, ITDRecord> itdRecords = itdDataLoader.loadITDRecords(symbolList, exchangeList
                    .getExchanges(exchanges), curDate, curDate);
            //Initialize index
            GlobalIndex index = createGlobalIndex(curDate, windows, itdRecords, windowSize);
            // Generate detailed match entry
            entryStream = Stream.concat(entryStream, symbolKeys.stream().map(k -> getMatchEntries(index, curDate,
                    itdRecords, k)).flatMap(e -> e.stream()));
        }
        return entryStream.collect(Collectors.toList());
    }

    private static List<MatchEntry> getMatchEntries(GlobalIndex index, LocalDate curDate, Multimap<String, ITDRecord>
            itdRecords, String searchSymbolKey) {
        Stopwatch stopwatch = Stopwatch.createStarted();
        // Search against each window -> result contains matched symbols for each window
        Map<LocalDateTime, Double> valueForSearch = getMidValueForSymbol(curDate, itdRecords, searchSymbolKey);
        Map<Window, List<String>> searchResult = index.search(valueForSearch, 0.0005);
        List<MatchEntry> matchEntries = new ArrayList<>();
        PearsonsCorrelation personsCorrelation = new PearsonsCorrelation();
        searchResult.entrySet().forEach(resEntry -> { // For each window
            Window window = resEntry.getKey();
            double[] baseVector = VectorUtil.flatToFirst(window.buildValue(valueForSearch));
            resEntry.getValue().stream().filter(symbol -> !symbol.equals(searchSymbolKey)).
                    forEach(symbol -> { // For each matched symbol
                        int vectorLength = baseVector.length;
                        double[] matchedVector = VectorUtil.flatToFirst(window.buildValue(getMidValueForSymbol
                                (curDate, itdRecords, symbol)));
                        // Calculate correlation between searchValue and matchedValue
                        double corr = personsCorrelation.correlation(baseVector, matchedVector);
                        double baseLastVal = baseVector[vectorLength - 1];
                        double matchLastVal = matchedVector[vectorLength - 1];
                        double diffLastVal = baseLastVal - matchLastVal;
                        double[] diff = VectorUtil.subtract(baseVector, matchedVector);
                        double positiveDiffRatio = (double) Arrays.stream(diff).filter(v -> v >= 0).count() /
                                (double) vectorLength;
                        double negativeDiffRatio = (double) Arrays.stream(diff).filter(v -> v <= 0).count() /
                                (double) vectorLength;
                        // Save match result to match entry object <window,symbolA,symbolB,correlation>
                        if (!Double.isNaN(corr)) {
                            matchEntries.add(new MatchEntry(window, searchSymbolKey, symbol, corr, baseLastVal,
                                    matchLastVal, diffLastVal, positiveDiffRatio, negativeDiffRatio));
                        }
                    });
        });
        logger.info("Found " + matchEntries.size() + " match entries.");
        logger.info("Search elapsed(s):" + stopwatch.elapsed(TimeUnit.MILLISECONDS));
        return matchEntries;
    }

    private static GlobalIndex createGlobalIndex(LocalDate curDate, List<Window> windows, Multimap<String, ITDRecord>
            itdRecords, int windowSize) {
        GlobalIndex index = new GlobalIndex(windows, windowSize);
        //Load mid-value and put into index
        Stopwatch stopwatch = Stopwatch.createStarted();
        itdRecords.keySet().stream().forEach(symbolKey -> {
            Map<LocalDateTime, Double> recordMapMid = getMidValueForSymbol(curDate, itdRecords, symbolKey);
            //Transform records to arrays for each window
            Map<Window, double[]> itdDaySource = windows.stream().
                    collect(Collectors.toMap(w -> w, w -> w.buildValue(recordMapMid)));
            itdDaySource.entrySet().stream().forEach(windowEntry -> {
                double[] dayEntryValue = windowEntry.getValue();
                if (dayEntryValue.length != 0 && Arrays.stream(dayEntryValue).
                        filter(v -> !Double.isNaN(v)).count() >= dayEntryValue.length * 1) {
                    double[] value = VectorUtil.flatToFirst(dayEntryValue);
                    index.insert(value, windowEntry.getKey(), symbolKey);
                }
            });
        });
        logger.info("Index creation elapsed(s):" + stopwatch.elapsed(TimeUnit.MILLISECONDS));
        return index;
    }

    private static Map<LocalDateTime, Double> getMidValueForSymbol(LocalDate curDate, Multimap<String, ITDRecord>
            itdRecords, String symbolKey) {
        Map<LocalDateTime, Double> resultMap = new TreeMap<>();
        itdRecords.get(symbolKey).stream().
                forEach(itdRecord -> resultMap.put(curDate.atTime(itdRecord.getDateTime().toLocalTime()), Math.sqrt
                        (itdRecord.getHigh() * itdRecord.getLow())));
        return resultMap;
    }
}
